package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;

import com.sunwayworld.basemodule.common.eptstorage.acqconfig.bean.LimsEquiptAcqConfigInstBean;
import com.sunwayworld.cloud.module.equipt.acqconfig.bean.CoreEquiptAcqConfigBean;
import com.sunwayworld.framework.exception.InvalidDataException;
import com.sunwayworld.framework.utils.ObjectUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.*;

/**
 * @auther zhangs
 * @description
 * @date 2022/3/5
 * 用于测试连接
 */
@Component
public class NetEptSenderTest {

    private static final Logger logger = LogManager.getLogger(NetEptSenderTest.class);
    //对于长期连不上的设备设置黑名单，重试3次连接不上，剔除
    public static final ConcurrentHashMap<String,NetEptSender> cacheMap = new ConcurrentHashMap<>(); //缓存设备连接

    public static void putNetEptSender(CoreEquiptAcqConfigBean acqEptBean,NetEptSender eptSender){
        if(acqEptBean!=null && eptSender!=null && eptSender.isConnect()){
            String key = getKey(acqEptBean);
            cacheMap.put(key,eptSender);
        }
    }
    public static void removeNetEptSender(CoreEquiptAcqConfigBean acqEptBean){
        if(acqEptBean!=null){
            String key = getKey(acqEptBean);
            cacheMap.remove(key);
        }
    }
    public NetEptSender getCacheNetEptSender(CoreEquiptAcqConfigBean acqEptBean){
        return cacheMap.get(getKey(acqEptBean));
    }


    private ThreadPoolExecutor.DiscardOldestPolicy rejectedExecutionHandler = new ThreadPoolExecutor.DiscardOldestPolicy(){
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            super.rejectedExecution(r, e);
            logger.warn("DiscardOldestPolicy.............");
        }
    };
    private  ThreadPoolExecutor connectExecutor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100),  Executors.defaultThreadFactory(), rejectedExecutionHandler);

    public void testConnect(CoreEquiptAcqConfigBean acqEptBean){
        int corePoolSize = connectExecutor.getCorePoolSize();
        int poolSize = connectExecutor.getPoolSize();
        int activeCount = connectExecutor.getActiveCount();
        long queueSize = connectExecutor.getQueue().size();
        logger.info("检测设备net连接 connectExecutor,corePoolSize={},poolSize={},activeCount={},taskCount={}",corePoolSize,poolSize,activeCount,queueSize);
        String key = acqEptBean.getIp()+":"+acqEptBean.getPort();
        NetEptSender netEptSender = cacheMap.get(key); //连接复用
        String netacqtype = acqEptBean.getExt$Item("NETACQTYPE");
        if(netEptSender==null || !netEptSender.isConnect()){
            logger.info("重新连接={},{}....",key,netacqtype);
            if(netEptSender!=null){
                netEptSender.close();
            }

            connectExecutor.submit(()->{
                NetEptSender newNetEptSender = null;
                if("tcp".equalsIgnoreCase(netacqtype)){
                    newNetEptSender = new NetEptTcpSender(acqEptBean);
                }else if("udp".equalsIgnoreCase(netacqtype)){
                    newNetEptSender = new NetEptUdpSender(acqEptBean);
                }
                newNetEptSender.getConnection(); //必须异步连接
            });
        }else{
            logger.info("复用连接={},{}....",key,netacqtype);
        }
    }

    //存储待发送的设备-指令队列 map
    @Transactional
    public synchronized void send(CoreEquiptAcqConfigBean acqEptBean, LimsEquiptAcqConfigInstBean instBean){
        if(acqEptBean==null || ObjectUtils.isEmpty(instBean)){
            return;
        }
        String key = getKey(acqEptBean);
        NetEptSender netEptSender = cacheMap.get(key);
        if(netEptSender==null){
            logger.error("netEptSender={} is null ",key);
            throw new InvalidDataException("连接失败！");
        }
        if(!netEptSender.isConnect()){
            logger.error("netEptSender={} is not connect ",key);
            throw new InvalidDataException("连接失败！");
        }
        netEptSender.send(instBean);
    }

    private static String getKey(CoreEquiptAcqConfigBean configBean){
        return configBean.getIp()+":"+configBean.getPort();
    }

}
